package org.ngbx.demo.flink.clickevent.mocker;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.ngbx.demo.flink.clickevent.UserClickedEvent;

import java.util.Arrays;
import java.util.Properties;

public class UserClickedEventSourceMocker {
    // 延迟：毫秒
    public static final long DELAY = 3000;

    public static void main(String[] args) throws Exception {
        // 创建kafka配置属性
        Properties kafkaProps = createKafkaProperties();

        // 创建Kafka消息的生产者
        KafkaProducer<String, byte[]> producer = new KafkaProducer<>(kafkaProps);

        String topic = "user_clicked_event";

        UserClickedEventIterator.userIdList = Arrays.asList(10000L, 10001L, 10002L, 10003L);
        UserClickedEventIterator clickIterator = new UserClickedEventIterator();

        // 循环发送事件
        while (true) {
            UserClickedEvent newEvent = clickIterator.next();
            System.out.println(newEvent);

            ProducerRecord<String, byte[]> record = new UserClickedEventSerialization(topic).serialize(
                    newEvent,
                    null);

            producer.send(record);

            Thread.sleep(DELAY);
        }
    }

    private static Properties createKafkaProperties() {
        Properties kafkaProps = new Properties();
        kafkaProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092");
        kafkaProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getCanonicalName());
        kafkaProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getCanonicalName());
        return kafkaProps;
    }
}
